package com.itc.bi.kafaka;


import com.alibaba.fastjson.JSONObject;
import com.itc.bi.constant.KafkaConstant;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;


/**
 * @program: itc-airport-cloud
 * @description: 监听mqtt消息
 * @author: songxiude
 * @create: 2024-07-17 10:44
 **/
@Slf4j
@Component
public class MqttDataKafkaListener {

    @Autowired
    private MqttDataAcceptService dataAcceptService;

    /**
     * 处理mqtt传来的数据
     * @param record
     */
    @KafkaListener(groupId = "mqttDataAccept",topics = KafkaConstant.MQTT_DATA_ACCEPT)
    public void mqttDataAccept(ConsumerRecord record){
        String message = (String) record.value();
        //JSONObject jsonObject = JSONObject.parseObject(message, JSONObject.class);
        //异步处理，避免kafka消息堆积
        dataAcceptService.dataAccept(message);

    }

}